package com.atguigu.chapter11;

import com.atguigu.chapter5.source.WaterSensor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;

import java.text.DecimalFormat;

/**
 * @ClassName: Flink14_Function_Scalar
 * @Description:
 * @Author: kele
 * @Date: 2021/4/14 0:09
 *
 *
 * 聚合函数：
 *     1、extends AggregateFunction
 *     2、需要实现3个方法：
 *          --获取最终结果：getValue，
 *          --初始化acc：createAccumulator，
 *          --存储中间变量：accumulate
 *
 **/
public class Flink16_Function__Aggregate {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

        DataStreamSource<WaterSensor> ds = env.fromElements(new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_2", 6000L, 60));

        Table table = tenv.fromDataStream(ds);

        /**
         * 内联方式使用
         */
     /*   table
                .groupBy($("id"))
                .select($("id"),call(MyAggFunc.class,$("vc")).as("vc_sum") )
                .execute()
                .print();*/


        /**
         * 注册后使用
         */

     /*   tenv.createTemporaryFunction("Myagg",MyAggFunc.class);

        table
                .groupBy($("id"))
                .select($("id"),call("Myagg",$("vc")))
                .execute()
                .print();*/


        /**
         * sql的方式使用
         */
        tenv.createTemporaryFunction("Myagg",MyAggFunc.class);
        tenv.createTemporaryView("mtable",table);

        tenv.executeSql("select id, Myagg(vc) vc_avg from mtable group by id")
                .print();


    }

    /**
     * 泛型 1：最终输出结果
     * 泛型 2：中间结果
     */
    public static class MyAggFunc extends AggregateFunction<Double, Tuple2<Long,Long>>{

        /**
         * 输出最终结果
         * @param acc
         * @return
         */
        @Override
        public Double getValue(Tuple2<Long, Long> acc) {

            if(acc.f1 == 0)
                return 0D;

            DecimalFormat df = new DecimalFormat("0.00");

            return Double.valueOf(df.format(acc.f0 * 1.0 / acc.f1));
        }

        /**
         * 创建累加器，初始化
         * @return
         */
        @Override
        public Tuple2<Long, Long> createAccumulator() {

            return Tuple2.of(0L,0L);

        }

        /**
         * 参数1：累加器的中间结果
         * 参数.... :  外界给的参数
         */
        public void accumulate(Tuple2<Long,Long> acc,Integer vc){

            if(vc != null){
                acc.f1 +=+ 1L;
                acc.f0 += Long.valueOf(vc);
            }
        }
    }

}
